package com.gzyj.flink.gps.map;

import com.gzyj.flink.dto.alarm.Kafka_VehicleAreaRiskAlarm;
import com.gzyj.flink.gps.GPSPosition;
import com.gzyj.flink.gps.map.area.AreaRiskAlarmDataProcess;
import com.gzyj.flink.gps.sink.gp.AreaRiskAlarmSink;
import com.gzyj.flink.pojo.gps.TsRiskAreaAlarm;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;

import java.util.Date;
import java.util.Objects;

public class AreaRiskAlarmFlow {

    private SingleOutputStreamOperator<GPSPosition> gpsSingleStream;

    public AreaRiskAlarmFlow(SingleOutputStreamOperator<GPSPosition> gpsSingleStream) {
        this.gpsSingleStream = gpsSingleStream;
    }

    public void process() {

        SingleOutputStreamOperator<Kafka_VehicleAreaRiskAlarm> areaRiskAlarmSingleOutputStream = this.gpsSingleStream.map(new AreaRiskAlarmDataProcess())
                .filter(Objects::nonNull)
                .filter(e->e.getDependencyId()!=null&&e.getVehicleId()!=null);
        areaRiskAlarmSingleOutputStream.map(new MapFunction<Kafka_VehicleAreaRiskAlarm, TsRiskAreaAlarm>() {
            @Override
            public TsRiskAreaAlarm map(Kafka_VehicleAreaRiskAlarm riskData) throws Exception {
                TsRiskAreaAlarm tsRiskAreaAlarm=new TsRiskAreaAlarm();
                tsRiskAreaAlarm.setVehId(riskData.getVehicleId());
                tsRiskAreaAlarm.setRiskAreaId(riskData.getRiskId());
                tsRiskAreaAlarm.setAreaId(riskData.getDependencyId());
                tsRiskAreaAlarm.setAreaName(riskData.getDependencyName());
                tsRiskAreaAlarm.setAreaIntime(riskData.getInDependencyTime());
                tsRiskAreaAlarm.setAreaInlocation(riskData.getInDependencyRoadName());
                tsRiskAreaAlarm.setAreaInlongitude(riskData.getInDependencyLongitude());
                tsRiskAreaAlarm.setAreaInlatitude(riskData.getInDependencyLatitude());
                tsRiskAreaAlarm.setRiskAreaName(riskData.getRiskName());
                tsRiskAreaAlarm.setRiskAreaIntime(riskData.getInRiskTime());
                tsRiskAreaAlarm.setRiskAreaInlocation(riskData.getInRiskRoadName());
                tsRiskAreaAlarm.setRiskAreaInlongitude(riskData.getInRiskLongitude());
                tsRiskAreaAlarm.setRiskAreaInlatitude(riskData.getInRiskLatitude());
                tsRiskAreaAlarm.setRiskAreaOuttime(riskData.getOutRiskTime());
                tsRiskAreaAlarm.setRiskAreaOutlocation(riskData.getOutRiskRoadName());
                tsRiskAreaAlarm.setRiskAreaOutlongitude(riskData.getOutRiskLongitude());
                tsRiskAreaAlarm.setRiskAreaOutlatitude(riskData.getOutRiskLatitude());
                tsRiskAreaAlarm.setAlarmType(riskData.getNote());
                tsRiskAreaAlarm.setAlarmTime(riskData.getInDependencyTime());
                tsRiskAreaAlarm.setCreateBy("admin");
                tsRiskAreaAlarm.setCreateTime(new Date());
                tsRiskAreaAlarm.setDriverName(riskData.getDriver());
                tsRiskAreaAlarm.setDriverYyz(riskData.getDriverNo());
                tsRiskAreaAlarm.setDriverNo(riskData.getDriverNo());
                return tsRiskAreaAlarm;
            }
        }).addSink(new AreaRiskAlarmSink());
    }


}
